package com.xrui.hbase.impl;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.base.Objects;

import com.xrui.hbase.*;
import com.xrui.hbase.codec.CellDecoder;
import com.xrui.hbase.codec.impl.CellDecoderProvider;
import com.xrui.hbase.compression.CompressionProvider;
import com.xrui.hbase.exception.ColumnPagingNotEnabledException;
import com.xrui.hbase.exception.InvalidNameException;
import com.xrui.hbase.exception.NoSuchColumnException;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.util.TimestampComparator;
import org.apache.avro.Schema;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.KVComparator;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;


/**
 * An implementation of RowData that wraps an HBase Result object.
 */

public final class HBaseRowData implements RowData {
    //region static
    private static final Logger LOG = LoggerFactory.getLogger(HBaseRowData.class);
    //endregion

    //region variables
    /**
     * The entity id for the row.
     */
    private final EntityId mEntityId;

    /**
     * The request used to retrieve this  row data.
     */
    private final DataRequest mDataRequest;

    /**
     * The HBase Table we are reading from.
     */
    private final HBaseTable mTable;

    /**
     * The layout for the table this row data came from.
     */
    private final TableLayout mTableLayout;
    /**
     * Provider for cell decoders.
     */
    private final CellDecoderProvider mDecoderProvider;
    /**
     * The HBase result providing the data of this object.
     */
    private Result mResult;
    /**
     * A map from  family to  qualifier to timestamp to raw encoded cell values.
     */
    private NavigableMap<String, NavigableMap<String, NavigableMap<Long, byte[]>>> mFilteredMap;
    //endregion

    //region construct

    /**
     * Initializes a row data from an HBase Result.
     * <p>
     * <p>
     * The HBase Result may contain more cells than are requested by the user.
     * DataRequest is more expressive than HBase Get/Scan requests.
     * Currently, {@link #getMap()} attempts to complete the filtering to meet the data request
     * requirements expressed by the user, but this can be inaccurate.
     * </p>
     *
     * @param table           table containing this row.
     * @param dataRequest     Data requested for this row.
     * @param entityId        This row entity ID.
     * @param result          HBase result containing the requested cells (and potentially more).
     * @param decoderProvider Provider for cell decoders.
     *                        Null means the row creates its own provider for cell decoders (not recommended).
     * @throws IOException on I/O error.
     */
    public HBaseRowData(
        final HBaseTable table,
        final DataRequest dataRequest,
        final EntityId entityId,
        final Result result,
        final CellDecoderProvider decoderProvider
    ) throws IOException {
        mTable = table;
        mTableLayout = table.getLayout();
        mDataRequest = dataRequest;
        mEntityId = entityId;
        mResult = result;
        mDecoderProvider = ((decoderProvider != null) ? decoderProvider : createCellProvider(table))
            .getDecoderProviderForRequest(dataRequest);
    }

    //endregion

    //region private methods

    /**
     * Creates a provider for cell decoders.
     * <p>
     * <p> The provider creates decoders for specific Avro records. </p>
     *
     * @param table HBase Table to create a CellDecoderProvider for.
     * @return a new CellDecoderProvider for the specified HBase Table.
     * @throws IOException on I/O error.
     */
    private static CellDecoderProvider createCellProvider(
        final HBaseTable table
    ) throws IOException {
        return CellDecoderProvider.create(
            table.getLayout(),
            ImmutableMap.<ColumnName, BoundColumnReaderSpec>of(),
            ImmutableList.<BoundColumnReaderSpec>of(),
            TableReaderBuilder.DEFAULT_CACHE_MISS);
    }

    /**
     * Reports the encoded map of qualifiers of a given family.
     *
     * @param family Family to look up.
     * @return the encoded map of qualifiers in the specified family, or null.
     */
    private NavigableMap<String, NavigableMap<Long, byte[]>> getRawQualifierMap(final String family) {
        return getMap().get(family);
    }

    /**
     * Reports the specified raw encoded time-series of a given column.
     *
     * @param family    Family to look up.
     * @param qualifier Qualifier to look up.
     * @return the encoded time-series in the specified family:qualifier column, or null.
     */
    private NavigableMap<Long, byte[]> getRawTimestampMap(
        final String family,
        final String qualifier
    ) {
        final NavigableMap<String, NavigableMap<Long, byte[]>> qmap = getRawQualifierMap(family);
        if (null == qmap) {
            return null;
        }
        return qmap.get(qualifier);
    }

    /**
     * Reports the encoded content of a given cell.
     *
     * @param family    Family to look up.
     * @param qualifier Qualifier to look up.
     * @param timestamp Timestamp to look up.
     * @return the encoded cell content, or null.
     */
    private byte[] getRawCell(String family, String qualifier, long timestamp) {
        final NavigableMap<Long, byte[]> tmap = getRawTimestampMap(family, qualifier);
        if (null == tmap) {
            return null;
        }
        return tmap.get(timestamp);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public EntityId getEntityId() {
        return mEntityId;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public RowKeyComponents getRowKeyComponents() throws IOException {
        return mTableLayout.getRowKeyDecoder().decode(
            mEntityId.getHBaseRowKey(),
            mTableLayout.getRowKeyFormat()
        );
    }

    /**
     * Gets the HBase result backing this {@link   RowData}.
     *
     * @return The HBase result.
     */
    public Result getHBaseResult() {
        return mResult;
    }

    //endregion

    //region APIs
    //region light weight gets based on queried result

    /**
     * Gets the table this row data belongs to.
     *
     * @return the table this row data belongs to.
     */
    public HBaseTable getTable() {
        return mTable;
    }

    /**
     * Gets the data request used to retrieve this row data.
     *
     * @return the data request used to retrieve this row data.
     */
    public DataRequest getDataRequest() {
        return mDataRequest;
    }

    /**
     * Gets the layout of the table this row data belongs to.
     *
     * @return The table layout.
     */
    public TableLayout getTableLayout() {
        return mTableLayout;
    }

    /**
     * Gets a map from family to qualifier to timestamp to raw encoded bytes of a cell.
     *
     * @return The map.
     */
    public synchronized NavigableMap<String, NavigableMap<String, NavigableMap<Long, byte[]>>>
    getMap() {
        if (null != mFilteredMap) {
            return mFilteredMap;
        }

        LOG.debug("Filtering the HBase Result into a map of cells...");
        final NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> map =
            mResult.getMap();
        mFilteredMap = new TreeMap<String, NavigableMap<String, NavigableMap<Long, byte[]>>>();
        if (null == map) {
            LOG.debug("No result data.");
            return mFilteredMap;
        }

        // Loop over the families in the HTable.
        for (NavigableMap.Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> familyEntry
            : map.entrySet()) {
            // Loop over the columns in the family.
            for (NavigableMap.Entry<byte[], NavigableMap<Long, byte[]>> columnEntry
                : familyEntry.getValue().entrySet()) {
                ColumnName columnName;
                try {
                    columnName = ColumnName.create(Bytes.toString(familyEntry.getKey()), Bytes.toString(columnEntry.getKey()));
                } catch (InvalidNameException e) {
                    LOG.info("Ignoring HBase column '{}:{}' because it doesn't contain  data.",
                        Bytes.toString(familyEntry.getKey()),
                        Bytes.toString(columnEntry.getKey()));
                    continue;
                }
                LOG.debug("Adding family [{}] to getMap() result.", columnName.getName());

                // First check if all columns were requested.
                DataRequest.Column columnRequest =
                    mDataRequest.getColumn(columnName.getFamily(), null);
                if (null == columnRequest) {
                    // Not all columns were requested, so check if this particular column was.
                    columnRequest =
                        mDataRequest.getColumn(columnName.getFamily(), columnName.getQualifier());
                }
                if (null == columnRequest) {
                    LOG.debug("Ignoring unrequested data: " + columnName.getFamily() + ":"
                        + columnName.getQualifier());
                    continue;
                }

                // Loop over the versions.
                int numVersions = 0;
                for (NavigableMap.Entry<Long, byte[]> versionEntry : columnEntry.getValue().entrySet()) {
                    if (numVersions >= columnRequest.getMaxVersions()) {
                        LOG.debug("Skipping remaining cells because we hit max versions requested: "
                            + columnRequest.getMaxVersions());
                        break;
                    }

                    // Read the timestamp.
                    final long timestamp = versionEntry.getKey();
                    if (mDataRequest.isTimestampInRange(timestamp)) {
                        // Add the cell to the filtered map.
                        if (!mFilteredMap.containsKey(columnName.getFamily())) {
                            mFilteredMap.put(columnName.getFamily(),
                                new TreeMap<String, NavigableMap<Long, byte[]>>());
                        }
                        final NavigableMap<String, NavigableMap<Long, byte[]>> columnMap =
                            mFilteredMap.get(columnName.getFamily());
                        if (!columnMap.containsKey(columnName.getQualifier())) {
                            columnMap.put(columnName.getQualifier(),
                                new TreeMap<Long, byte[]>(TimestampComparator.INSTANCE));
                        }
                        final NavigableMap<Long, byte[]> versionMap =
                            columnMap.get(columnName.getQualifier());
                        versionMap.put(versionEntry.getKey(), versionEntry.getValue());
                        ++numVersions;
                    } else {
                        LOG.debug("Excluding cell at timestamp " + timestamp + " because it is out of range ["
                            + mDataRequest.getMinTimestamp() + "," + mDataRequest.getMaxTimestamp() + ")");
                    }
                }
            }
        }
        return mFilteredMap;
    }

    public NavigableMap<String, NavigableMap<String, NavigableMap<Long, byte[]>>>
    getMap(String family, String qualifier) throws IOException {
        if (null == family) {
            return getMap();
        }

        if (null == qualifier) {
            final NavigableMap<String, NavigableMap<Long, byte[]>> qualifierMap = getRawQualifierMap(family);
            if (null == qualifierMap) {
                return null;
            }

            NavigableMap<String, NavigableMap<String, NavigableMap<Long, byte[]>>> ret = new TreeMap<>();
            ret.put(family, qualifierMap);
            return ret;
        }

        NavigableMap<Long, byte[]> timeStampMap = getRawTimestampMap(family, qualifier);
        final NavigableMap<String, NavigableMap<Long, byte[]>> qualifierMap = new TreeMap<>();
        final NavigableMap<String, NavigableMap<String, NavigableMap<Long, byte[]>>> ret = new TreeMap<>();
        qualifierMap.put(qualifier, timeStampMap);
        ret.put(family, qualifierMap);

        return ret;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized NavigableSet<String> getQualifiers(final String family) {
        final NavigableMap<String, NavigableMap<Long, byte[]>> qmap = getRawQualifierMap(family);
        if (null == qmap) {
            return Sets.newTreeSet();
        }
        return qmap.navigableKeySet();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized NavigableSet<Long> getTimestamps(
        final String family,
        final String qualifier
    ) {
        final NavigableMap<Long, byte[]> tmap = getRawTimestampMap(family, qualifier);
        if (null == tmap) {
            return Sets.newTreeSet(TimestampComparator.INSTANCE);
        }
        return tmap.navigableKeySet();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> T getMostRecentValue(String family, String qualifier) throws IOException {
        final HBaseDecoderWrapper decoderWrapper = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout,
            family, qualifier);
        final NavigableMap<Long, byte[]> tmap = getRawTimestampMap(family, qualifier);
        if (null == tmap) {
            return null;
        }
        final byte[] bytes = tmap.values().iterator().next();
        return (T) decoderWrapper.decodeValue(bytes);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> NavigableMap<String, T> getMostRecentValues(String family) throws IOException {
        //Preconditions.checkState(mTableLayout.getFamilyMap().create(family).isMapType(),
        //    "getMostRecentValues(String family) is only enabled"
        //        + " on map type column families. The column family [%s], is a group type column family."
        //        + " Please use the getMostRecentValues(String family, String qualifier) method.",
        //    family);
        final NavigableMap<String, T> result = Maps.newTreeMap();
        for (String qualifier : getQualifiers(family)) {
            final T value = getMostRecentValue(family, qualifier);
            result.put(qualifier, value);
        }
        return result;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> NavigableMap<Long, T> getValues(String family, String qualifier)
        throws IOException {
        final NavigableMap<Long, T> result = Maps.newTreeMap(TimestampComparator.INSTANCE);
        for (Map.Entry<Long, HBaseCell<T>> entry : this.<T>getCells(family, qualifier).entrySet()) {
            result.put(entry.getKey(), entry.getValue().getData());
        }
        return result;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> NavigableMap<String, NavigableMap<Long, T>> getValues(String family)
        throws IOException {
        //Preconditions.checkState(mTableLayout.getFamilyMap().create(family).isMapType(),
        //    "getValues(String family) is only enabled on map "
        //        + "type column families. The column family [%s], is a group type column family. Please use "
        //        + "the getValues(String family, String qualifier) method.",
        //    family);
        final NavigableMap<String, NavigableMap<Long, T>> result = Maps.newTreeMap();
        for (String qualifier : getQualifiers(family)) {
            final NavigableMap<Long, T> timeseries = getValues(family, qualifier);
            result.put(qualifier, timeseries);
        }
        return result;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> T getValue(String family, String qualifier, long timestamp) throws IOException {
        final HBaseDecoderWrapper decoderWrapper = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout,
            family, qualifier);

        final byte[] bytes = getRawCell(family, qualifier, timestamp);
        if (null == bytes) {
            return null;
        }
        return (T) decoderWrapper.decodeValue(bytes);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> HBaseCell<T> getCell(String family, String qualifier, long timestamp)
        throws IOException {
        final HBaseDecoderWrapper decoderWrapper = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout,
            family, qualifier);
        final byte[] bytes = getRawCell(family, qualifier, timestamp);
        if (null == bytes) {
            return null;
        }
        return decoderWrapper.decodeHBaseCell(mEntityId, timestamp, bytes);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> HBaseCell<T> getMostRecentCell(String family, String qualifier) throws IOException {
        final HBaseDecoderWrapper decoderWrapper = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout,
            family, qualifier);

        final NavigableMap<Long, byte[]> tmap = getRawTimestampMap(family, qualifier);
        if (null == tmap) {
            return null;
        }
        final byte[] bytes = tmap.values().iterator().next();
        final long timestamp = tmap.firstKey();

        return decoderWrapper.decodeHBaseCell(mEntityId, timestamp, bytes);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> NavigableMap<String, HBaseCell<T>> getMostRecentCells(String family)
        throws IOException {
        //Preconditions.checkState(mTableLayout.getFamilyMap().create(family).isMapType(),
        //    "getMostRecentCells(String family) is only enabled"
        //        + " on map type column families. The column family [%s], is a group type column family."
        //        + " Please use the getMostRecentCells(String family, String qualifier) method.",
        //    family);
        final NavigableMap<String, HBaseCell<T>> result = Maps.newTreeMap();
        for (String qualifier : getQualifiers(family)) {
            final HBaseCell<T> cell = getMostRecentCell(family, qualifier);
            result.put(qualifier, cell);
        }
        return result;
    }

    public <T> List<HBaseCell<T>> getCellList() throws IOException {
        List<HBaseCell<T>> result = new ArrayList<HBaseCell<T>>();

        final NavigableMap<String, NavigableMap<String, NavigableMap<Long, byte[]>>> resultMap = getMap();

        for (NavigableMap.Entry<String, NavigableMap<String, NavigableMap<Long, byte[]>>> fTree : resultMap.entrySet()) {
            String famliyValue = fTree.getKey();

            for (NavigableMap.Entry<String, NavigableMap<Long, byte[]>> qTree : fTree.getValue().entrySet()) {
                String qualifierValue = qTree.getKey();
                final HBaseDecoderWrapper decoderWrapper = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout,
                    famliyValue, qualifierValue);

                for (NavigableMap.Entry<Long, byte[]> tTree : qTree.getValue().entrySet()) {
                    long timestampValue = tTree.getKey();
                    final byte[] bytes = tTree.getValue();

                    final HBaseCell<T> cell = decoderWrapper.decodeHBaseCell(mEntityId, timestampValue, bytes);
                    result.add(cell);
                }
            }
        }
        return result;
    }

    public <T> List<HBaseCell<T>> getCellList(String family) throws IOException {
        if (null == family) {
            return getCellList();
        }

        List<HBaseCell<T>> result = new ArrayList<HBaseCell<T>>();

        final NavigableMap<String, NavigableMap<Long, byte[]>> fTree = getMap().get(family);
        if (null == fTree) {
            return result;
        }

        for (NavigableMap.Entry<String, NavigableMap<Long, byte[]>> qTree : fTree.entrySet()) {
            String qualifierValue = qTree.getKey();
            final HBaseDecoderWrapper decoderWrapper = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout,
                family, qualifierValue);

            for (NavigableMap.Entry<Long, byte[]> tTree : qTree.getValue().entrySet()) {
                long timestampValue = tTree.getKey();
                final byte[] bytes = tTree.getValue();

                final HBaseCell<T> cell = decoderWrapper.decodeHBaseCell(mEntityId, timestampValue, bytes);
                result.add(cell);
            }
        }

        return result;
    }

    public <T> List<HBaseCell<T>> getCellList(String family, String qualifier) throws IOException {
        if (null == family) {
            return getCellList();
        }

        if (null == qualifier) {
            return getCellList(family);
        }

        List<HBaseCell<T>> result = new ArrayList<HBaseCell<T>>();

        final NavigableMap<String, NavigableMap<Long, byte[]>> fTree = getMap().get(family);
        if (null == fTree) {
            return result;
        }

        final NavigableMap<Long, byte[]> qTree = fTree.get(qualifier);
        if (null == qTree) {
            return result;
        }

        final HBaseDecoderWrapper decoderWrapper = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout,
            family, qualifier);

        for (NavigableMap.Entry<Long, byte[]> tTree : qTree.entrySet()) {
            long timestampValue = tTree.getKey();
            final byte[] bytes = tTree.getValue();

            final HBaseCell<T> cell = decoderWrapper.decodeHBaseCell(mEntityId, timestampValue, bytes);
            result.add(cell);
        }

        return result;

    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> NavigableMap<Long, HBaseCell<T>> getCells(String family, String qualifier)
        throws IOException {
        final HBaseDecoderWrapper decoderWrapper = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout,
            family, qualifier);

        final NavigableMap<Long, HBaseCell<T>> result = Maps.newTreeMap(TimestampComparator.INSTANCE);
        final NavigableMap<Long, byte[]> tmap = getRawTimestampMap(family, qualifier);
        if (tmap != null) {
            for (Map.Entry<Long, byte[]> entry : tmap.entrySet()) {

                final Long timestamp = entry.getKey();
                final byte[] bytes = entry.getValue();
                final HBaseCell<T> cell = decoderWrapper.decodeHBaseCell(mEntityId, timestamp, bytes);

                result.put(timestamp, cell);
            }
        }
        return result;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> NavigableMap<String, NavigableMap<Long, HBaseCell<T>>> getCells(String family)
        throws IOException {
        //Preconditions.checkState(mTableLayout.getFamilyMap().create(family).isMapType(),
        //    "getCells(String family) is only enabled"
        //        + " on map type column families. The column family [%s], is a group type column family."
        //        + " Please use the getCells(String family, String qualifier) method.",
        //    family);
        final NavigableMap<String, NavigableMap<Long, HBaseCell<T>>> result = Maps.newTreeMap();
        for (String qualifier : getQualifiers(family)) {
            final NavigableMap<Long, HBaseCell<T>> cells = getCells(family, qualifier);
            result.put(qualifier, cells);
        }
        return result;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Pager getPager(String family, String qualifier)
        throws ColumnPagingNotEnabledException {
        final ColumnName columnName = ColumnName.create(family, qualifier);
        return new HBaseVersionPager(
            mEntityId, mDataRequest, mTable, columnName, mDecoderProvider);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Pager getPager(String family) throws ColumnPagingNotEnabledException {
        final ColumnName hbaseFamily = ColumnName.create(family, null);
        //Preconditions.checkState(mTableLayout.getFamilyMap().create(family).isMapType(),
        //    "getPager(String family) is only enabled on map type column families. "
        //        + "The column family '%s' is a group type column family. "
        //        + "Please use the getPager(String family, String qualifier) method.",
        //    family);
        return new HBaseMapFamilyPager(mEntityId, mDataRequest, mTable, hbaseFamily);
    }

    //endregion

    //region getPager for more data

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized boolean containsColumn(final String family, final String qualifier) {
        final NavigableMap<String, NavigableMap<Long, byte[]>> columnMap = getMap().get(family);
        if (null == columnMap) {
            return false;
        }
        final NavigableMap<Long, byte[]> versionMap = columnMap.get(qualifier);
        if (null == versionMap) {
            return false;
        }
        return !versionMap.isEmpty();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized boolean containsColumn(final String family) {

        final NavigableMap<String, NavigableMap<Long, byte[]>> columnMap = getMap().get(family);
        if (null == columnMap) {
            return false;
        }
        for (Map.Entry<String, NavigableMap<String, NavigableMap<Long, byte[]>>> columnMapEntry
            : getMap().entrySet()) {
            LOG.debug("The result return contains family [{}]", columnMapEntry.getKey());
        }
        return !columnMap.isEmpty();
    }

    //endregion

    //region contains

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized boolean containsCell(
        final String family,
        final String qualifier,
        long timestamp
    ) {
        return containsColumn(family, qualifier)
            && getTimestamps(family, qualifier).contains(timestamp);
    }
    //endregion

    //region Iterators
    //region for the specified column.

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> Iterator<HBaseCell<T>> iterator(String family, String qualifier)
        throws IOException {
        final ColumnName column = ColumnName.create(family, qualifier);
        Preconditions.checkArgument(
            mDataRequest.getRequestForColumn(column) != null,
            "Column %s has no data request.", column);
        return new CellIterator<T>(this, column);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> Iterable<HBaseCell<T>> asIterable(String family, String qualifier) {
        final ColumnName column = ColumnName.create(family, qualifier);
        Preconditions.checkArgument(
            mDataRequest.getRequestForColumn(column) != null,
            "Column %s has no data request.", column);
        return new CellIterable<T>(column, this);
    }

    //endregion

    //region for the specified family.

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> Iterator<HBaseCell<T>> iterator(String family)
        throws IOException {
        final ColumnName column = ColumnName.create(family, null);
        Preconditions.checkArgument(
            mDataRequest.getRequestForColumn(column) != null,
            "Column %s has no data request.", column);
        //Preconditions.checkState(mTableLayout.getFamilyMap().create(family).isMapType(),
        //    "iterator(String family) is only enabled"
        //        + " on map type column families. The column family [%s], is a group type column family."
        //        + " Please use the iterator(String family, String qualifier) method.",
        //    family);
        return new CellIterator<T>(this, column);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> Iterable<HBaseCell<T>> asIterable(String family) {
        final ColumnName column = ColumnName.create(family, null);
        Preconditions.checkArgument(
            mDataRequest.getRequestForColumn(column) != null,
            "Column %s has no data request.", column);
        //Preconditions.checkState(mTableLayout.getFamilyMap().create(family).isMapType(),
        //    "asIterable(String family) is only enabled"
        //        + " on map type column families. The column family [%s], is a group type column family."
        //        + " Please use the asIterable(String family, String qualifier) method.",
        //    family);
        return new CellIterable<T>(column, this);
    }
    //endregion
    //endregion

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        if (null != mFilteredMap) {
            return Objects.toStringHelper(HBaseRowData.class)
                .add("table", mTable.getURI())
                .add("entityId", getEntityId())
                .add("dataRequest", mDataRequest)
                .add("resultSize", mResult.size())
                .add("result", mResult)
                .add("map", getMap())
                .toString();
        }

        return Objects.toStringHelper(HBaseRowData.class)
            .add("table", mTable.getURI())
            .add("entityId", getEntityId())
            .add("dataRequest", mDataRequest)
            .add("resultSize", mResult.size())
            .add("result", mResult)
            //.add("map", getMap())
            .toString();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Schema getReaderSchema(final String family, final String qualifier) throws IOException {
        return mTableLayout.getCellSpec(ColumnName.create(family, qualifier)).getAvroSchema();
    }
    //endregion

    /**
     * An iterator for cells in group type column or map type column family.
     *
     * @param <T> The type parameter for the Cells being iterated over.
     */
    private static final class CellIterator<T> implements Iterator<HBaseCell<T>> {

        private static final KVComparator KV_COMPARATOR = new KVComparator();
        private final KeyValue[] mKeyValues;
        private final EntityId mEntityId;
        private final ColumnName mColumn;
        private final CellDecoderProvider mDecoderProvider;
        private final TableLayout mTableLayout;
        private final DataRequest mDataRequest;
        private int mMaxVersions;
        private int mCurrentVersions = 0;
        private int mNextIndex = 0;
        private HBaseCell<T> mNextCell = null;

        /**
         * Create a new CellIterator.
         *
         * @param rowData    RowData from which to retrieve cells.
         * @param columnName Column across which to iterate. May be a fully qualified column or map type
         *                   family.
         * @throws IOException In case of an error initializing the Iterator.
         */
        private CellIterator(
            final HBaseRowData rowData,
            final ColumnName columnName
        ) throws IOException {
            mKeyValues = /*SchemaPlatformBridge.create().keyValuesFromResult(rowData.mResult);*/rowData.mResult.raw();
            mEntityId = rowData.mEntityId;
            mColumn = columnName;
            mDataRequest = rowData.mDataRequest;
            mMaxVersions = rowData.mDataRequest.getRequestForColumn(mColumn).getMaxVersions();
            mDecoderProvider = rowData.mDecoderProvider;
            mTableLayout = rowData.getTableLayout();
            mNextIndex = findStartIndex();
            mNextCell = getNextCell();
        }

        /**
         * Finds the insertion point of the pivot KeyValue in the KeyValue array and returns the index.
         *
         * @param kvs           The KeyValue array to search in.
         * @param pivotKeyValue A KeyValue that is less than or equal to the first KeyValue for our
         *                      column, and larger than any KeyValue that may preceed values for our desired column.
         * @return The index of the first KeyValue in the desired map type family.
         */
        private static int findInsertionPoint(final KeyValue[] kvs, final KeyValue pivotKeyValue) {
            // Now find where the pivotKeyValue would be placed
            int binaryResult = Arrays.binarySearch(kvs, pivotKeyValue, KV_COMPARATOR);
            if (binaryResult < 0) {
                return -1 - binaryResult; // Algebra on the formula provided in the binary search JavaDoc.
            } else {
                return binaryResult;
            }
        }

        /**
         * Find the start index of the configured column in the KeyValues of this Iterator.
         *
         * @return the start index of the configured column in the KeyValues of this Iterator.
         * @throws NoSuchColumnException in case the column does not exist in the table.
         */
        private int findStartIndex() throws NoSuchColumnException {
            final KeyValue kv = new KeyValue(
                mEntityId.getHBaseRowKey(),
                mColumn.getFamilyBytes(),
                mColumn.getQualifierBytes(),
                Long.MAX_VALUE,
                new byte[0]);
            return findInsertionPoint(mKeyValues, kv);
        }

        /**
         * Get the next cell to be returned by this Iterator. Null indicates the iterator is exhausted.
         *
         * @return the next cell to be returned by this Iterator or null if the iterator is exhausted.
         * @throws IOException in case of an error decoding the cell.
         */
        private HBaseCell<T> getNextCell() throws IOException {
            // Ensure that we do not attempt to create KeyValues from out of bounds indices.
            if (mNextIndex >= mKeyValues.length) {
                return null;
            }

            final KeyValue next = mKeyValues[mNextIndex];
            final ColumnName column = ColumnName.create(
                Bytes.toString(next.getFamily()),
                Bytes.toString(next.getQualifier()));

            // Validates that the column of the next KeyValue should be included in the iterator.
            if (mColumn.isFullyQualified()) {
                if (!Objects.equal(mColumn, column)) {
                    // The column of the next cell is not the requested column, do not return it.
                    return null;
                }
            } else {
                if (!Objects.equal(column.getFamily(), mColumn.getFamily())) {
                    // The column of the next cell is not in the requested family, do not return it.
                    return null;
                }
                mMaxVersions = mDataRequest.getRequestForColumn(column).getMaxVersions();
            }

            if ((null != mNextCell) && !Objects.equal(mNextCell.getColumn(), column)) {
                // We've hit the next qualifier before the max versions; reset the current version count.
                mCurrentVersions = 0;
            }

            if (mCurrentVersions < mMaxVersions) {
                // decode the cell and return it.
                mCurrentVersions++;
                mNextIndex++;

                HBaseDecoderWrapper decoder = HBaseDecoderWrapper.create(mDecoderProvider, mTableLayout, column);
                return decoder.decodeHBaseCell(mEntityId, next.getTimestamp(), next.getValue());
            } else {
                // Reset the current versions and try the next qualifier.
                mCurrentVersions = 0;
                final KeyValue nextQualifierKV = new KeyValue(
                    mEntityId.getHBaseRowKey(),
                    column.getFamilyBytes(),
                    Arrays.copyOf(column.getQualifierBytes(), column.getQualifierBytes().length + 1),
                    Long.MAX_VALUE,
                    new byte[0]);
                mNextIndex = findInsertionPoint(mKeyValues, nextQualifierKV);
                return getNextCell();
            }
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public boolean hasNext() {
            return (null != mNextCell);
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public HBaseCell<T> next() {
            final HBaseCell<T> next = mNextCell;
            if (null == next) {
                throw new NoSuchElementException();
            } else {
                try {
                    mNextCell = getNextCell();
                } catch (IOException ioe) {
                    throw new com.xrui.hbase.exception.IOException(ioe);
                }
                return next;
            }
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public void remove() {
            throw new UnsupportedOperationException(
                String.format("%s does not support remove().", getClass().getName()));
        }
    }

    /**
     * An iterable of cells in a column.
     *
     * @param <T> The type parameter for the Cells being iterated over.
     */
    private static final class CellIterable<T> implements Iterable<HBaseCell<T>> {
        /**
         * The column family.
         */
        private final ColumnName mColumnName;
        /**
         * The rowdata we are iterating over.
         */
        private final HBaseRowData mRowData;

        /**
         * An iterable of Cells, for a particular column.
         *
         * @param colName The  column family that is being iterated over.
         * @param rowdata The HBaseRowData instance containing the desired data.
         */
        protected CellIterable(
            final ColumnName colName,
            final HBaseRowData rowdata
        ) {
            mColumnName = colName;
            mRowData = rowdata;
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public Iterator<HBaseCell<T>> iterator() {
            try {
                return new CellIterator<T>(
                    mRowData,
                    mColumnName);
            } catch (IOException ex) {
                throw new com.xrui.hbase.exception.IOException(ex);
            }
        }
    }

}

class HBaseDecoderWrapper<T> {
    //region static
    //cache the compressionProvider to make the query for each cell faster
    private static Map<String, CompressionProvider> mCompressionCache = new HashMap<>();
    //endregion

    //region variables
    private final CellDecoder<T> mdecoder;
    private final ColumnName mColumnName;
    private CompressionProvider mCompressionProvider;
    //endregion

    //region private construct
    private HBaseDecoderWrapper(CellDecoderProvider decoderProvider, TableLayout tableLayout, ColumnName columnName)
        throws IOException {
        mColumnName = columnName;
        mdecoder = decoderProvider.getDecoder(mColumnName);

        final String compressionName = tableLayout.getCellSchema(mColumnName).getCompression();
        mCompressionProvider = mCompressionCache.get(compressionName);
        if (null == mCompressionProvider) {
            mCompressionProvider = CompressionProvider.Factory.create(compressionName);
            mCompressionCache.put(compressionName, mCompressionProvider);
        }

    }
    //endregion

    //region static create methods
    public static HBaseDecoderWrapper create(CellDecoderProvider decoderProvider, TableLayout tableLayout,
                                             ColumnName columnName) throws IOException {
        return new HBaseDecoderWrapper(decoderProvider, tableLayout, columnName);
    }

    public static HBaseDecoderWrapper create(CellDecoderProvider decoderProvider, TableLayout tableLayout,
                                             String family, String qualifier) throws IOException {
        return new HBaseDecoderWrapper(decoderProvider, tableLayout, ColumnName.create(family, qualifier));
    }
    //endregion

    //region APIs
    public T decodeValue(byte[] bytes) throws IOException {
        return mdecoder.decodeValue(mCompressionProvider.decompress(bytes));
    }

    public HBaseCell<T> decodeHBaseCell(EntityId entityId, long timestamp, byte[] value) throws IOException {
        DecodedCell<T> decodedCell = mdecoder.decodeCell(mCompressionProvider.decompress(value));

        return HBaseCell.create(entityId, mColumnName, timestamp, decodedCell);
    }
    //endregion
}
